package com.xuef.dxf.namenode.server;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 管理集群中所有节点
 */
public class DataNodeManager {
    /**
     * 集群中所有节点
     */
    Map<String, DataNodeInfo> dataNodeInfoMap = new ConcurrentHashMap<>();

    public boolean registerDataNode(DataNodeInfo dataNodeInfo){
        dataNodeInfoMap.put(dataNodeInfo.getIp()+"-"+dataNodeInfo.getHostname(), dataNodeInfo);
        System.out.println("registered data node: " + dataNodeInfo.getIp()+"-"+dataNodeInfo.getHostname());
        return true;
    }

    /**
     *  DataNode heartbeat 过来，更新节点的心跳信息
     * @param dataNodeInfo
     */
    public void heartbeat(DataNodeInfo dataNodeInfo) {
        DataNodeInfo nodeInfo = dataNodeInfoMap.get(dataNodeInfo.getIp() + "-" + dataNodeInfo.getHostname());
        if(nodeInfo != null){
            nodeInfo.setLatestHeartbeatTime(System.currentTimeMillis());
            System.out.println("heartbeat from data node: " + dataNodeInfo.getIp()+"-"+dataNodeInfo.getHostname());
        }
    }

    public void startHeartbeatMonitor() {
        new DataNodeAliveMonitorThread().start();
    }

    /**
     * DataNode节点心跳监控线程
     */
    class DataNodeAliveMonitorThread extends Thread{
        @Override
        public void run() {
            while(true){
                System.out.println("data nodes heat beat monitor.......");
                List<String> toRemoveNodes = new ArrayList<>();
                //TODO DataNode心跳检测 将死的节点从列表删除
                Iterator<DataNodeInfo> dataNodeInfoIterator = dataNodeInfoMap.values().iterator();
                System.out.println("dataNodeInfoMap: " + dataNodeInfoMap);
                while(dataNodeInfoIterator.hasNext()){
                    DataNodeInfo next = dataNodeInfoIterator.next();
                    //已经断开心跳 30s
                    if(next.getLatestHeartbeatTime() + 30*1000 < System.currentTimeMillis()){
                        toRemoveNodes.add(next.getIp() + "-" + next.getHostname());
                    }
                }
                //删除死节点
                if(!toRemoveNodes.isEmpty()){
                    for(String nodeKey:toRemoveNodes){
                        dataNodeInfoMap.remove(nodeKey);
                        System.out.println("leaved data node: " + nodeKey);
                    }
                }
                try {
                    //10s 检测一次
                    Thread.sleep(10*1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
